#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <iostream>
#include <pthread.h>

#include "hbase_client.h"
#include "perf.h"
#include "base.h"
#include "error_code.h"

#define COL_CLUSTER              "col"
#define HBASE_SERVER             "127.0.0.1"
#define HBASE_PORT               10000

//TestConf *tf;
//SafeQueue *sq;

int
print_result(RowResult *table)
{
    int count = 0;
    RowResult *tmp;
    
    tmp = table;
    while (tmp) {
        std::cout << "table: " << tmp->table << endl;
        std::cout << "row: " << tmp->row << endl;
        ColumnValue *cv;
        while (cv = tmp->nextcolumn()) {
            //printf("column: %s\n", cv->column);
            //printf("value: %s\n", cv->value);
            std::cout << "column: " << cv->column << endl;
            std::cout << "value: " << cv->value << endl;
        }
        count++;
        tmp = tmp->next;
        printf("-------------------------------next row\n");
    }
    printf("-------------------------------total row: %d\n", count);
    return 0;
}

Titem*
tlist_init()
{
    Titem *head;
    head = (Titem *)zz_malloc(sizeof(Titem));
    if (head == NULL) {
        DERROR("can not molloc Titem struct\n");
        return NULL;
    }
    memset(head, 0x0, sizeof(Titem));
    
    return head;
}

int
tlist_insert(Titem *head, long long timediff)
{
    Titem *item;

    item = (Titem *)zz_malloc(sizeof(Titem));
    if (item == NULL) {
        DERROR("can not malloc Titem struct\n");
        return -1;
    }
    memset(item, 0x0, sizeof(Titem));
    item->timediff = timediff;
    
    item->next = head->next;
    head->next = item;
    return 0;
}

int
tlist_write(Titem *head)
{
    char file[1024];
    FILE *fptr = NULL;

    snprintf(file, 1024, "record/timfdiff_%u", pthread_self());
    
    fptr = fopen(file, "w+");
    if (fptr == NULL) {
        DERROR("can not open file: %s\n", file);
        return -1;
    }

    char buf[1024];
    Titem *item = head->next;
    while (item) {
        snprintf(buf, 1024,  "--------------thead %u quest time %d\n", pthread_self(), item->timediff);  
        fwrite(buf, strlen(buf), 1, fptr);
        item = item->next;
    }

    fclose(fptr);
    return 0;
}

int
tlist_destroy(Titem *head)
{
    if (head == NULL)
        return 0;
    Titem *item = head->next;
    Titem *tmp;

    while (item) {
        tmp = item;
        head = item->next;
        zz_free(tmp);
    }
    zz_free(head);
    return 0;
}

HbaseOp 
*hbase_connect()
{
    int ret;
    HbaseOp *ho;
    
    ho = new HbaseOp(HBASE_SERVER, HBASE_PORT);
    ret = ho->open();

    if (ret < 0)
        return NULL;
    return ho;
}

void
hbase_close(HbaseOp *op)
{
    int ret;
    
    ret = op->close();

    delete op;
}

int
insert()
{
    char prefix[10];
    char *value = NULL;
    int in = 'a';
    int count = 0;
    int ret;
    HbaseOp *ho;
    Titem *head;
    struct timeval start, end;
    struct timeval tstart, tend;
    
    ho = hbase_connect();
    if (ho == NULL) {
        printf("hbase connect error\n");
        return -1;
    }
    
    head = tlist_init(); 
    if (head == NULL) {
        return -1;
    }

    value = (char *)zz_malloc(tf->itemsize + 1);
    if (value == NULL) {
        DERROR("malloc error\n");
        return -1;
    }
    memset(value, in, tf->itemsize);
    *(value + tf->itemsize) = '\0';
    
    snprintf(prefix, 10, "%s:", COL_CLUSTER);

    int i;
    gettimeofday(&tstart, NULL);
    for (i = 0; i < tf->dcount_per_producer; i++) {
        struct RowResult *table = NULL;
        int j;
        for (j = 0; j < tf->lines_per_data; j++) {
            struct RowResult *row = new (struct RowResult);
            TableItem *ptr = tf->head;
            char buf[1024];
            int k = 0;
            while (ptr) {
                string col;
                col += prefix;
                col += ptr->name;

                if (k == 0) {
                    printf("========count: %d\n", count);
                    snprintf(buf, 1024, "%u_%010d", pthread_self(), count);
                    row->row = buf;
                    row->insert(col, buf, 0);
                } else {
                    row->insert(col, value, 0); 
                }
                ptr = ptr->next;
                k++;
            }
            count++;
            row->table = tf->tname;
            row->next = table;
            table = row;
        }
        //do something
        print_result(table);
        gettimeofday(&start, NULL);
        ret = ho->put(tf->tname, table);
        if (ret < 0 ) {
            if (ho->errnum == ERROR_CODE_SOCKET_CLOSED) {
                hbase_close(ho);
                ho = hbase_connect();
                if (NULL == ho) {
                    DERROR("can not reconnect hbase\n");
                    return -1;
                }
                ret = ho->put(tf->tname, table);
                if (ret < 0) {
                    DERROR("reconnect hbase, but can not put data\n");
                    return -1;
                }
            } else {
                hbase_close(ho);
                DERROR("error ocur, the end\n");
                return -1;
            }
        }
        gettimeofday(&end, NULL);
        long long diff = timediff(&start, &end);

        tlist_insert(head, diff);

        free_rows(table);
    }
    gettimeofday(&tend, NULL);

    printf("thread dispos time: %lld\n", timediff(&tstart, &tend));
    tlist_write(head);
    tlist_destroy(head);
    hbase_close(ho);    
    zz_free(value);
    printf("count: %d\n", count);
    return 0;
}

